// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: chensili

#pragma once

#include <map>
#include <memory>
#include <mutex>
#include <set>
#include <string>
#include <tuple>
#include <unordered_map>
#include <utility>

#include <boost/asio/ip/address.hpp>

#include "coin/proto/coin_query.pb.h"
#include "coin/proto/coin_request.pb.h"
#include "coin/proto/coin_executor.pb.h"
#include "coin2/strategy/order_executor.pb.h"
#include "coin2/exchange/base/log/health_reporter.h"
#include "coin2/exchange/base/log/strategy_logger.h"


namespace coin2::strategy::util {

using coin::proto::AccountRequestProto;
using coin2::exchange::base::strategy_util::PrometheusHealthReporter;
using coin2::exchange::base::strategy_util::StrategyLogger;
using coin2::exchange::base::strategy_util::StrategySlimLogger;
using coin2::exchange::base::strategy_util::TelemetryLogger;
using coin2::exchange::base::symbology::IProduct;

namespace {

unsigned short GetAvailablePort(unsigned short from_port, unsigned short to_port) {
  bool port_available = false;
  while (from_port < to_port) {
    boost::asio::io_service svc;
    boost::asio::ip::tcp::acceptor acceptor(svc);

    boost::system::error_code ec;
    acceptor.open(boost::asio::ip::tcp::v4(), ec) ||
                  acceptor.bind({boost::asio::ip::tcp::v4(), from_port }, ec);

    if (!ec) {
      port_available = true;
      break;
    }
    SPDLOG_INFO("Fail to use port: {}", from_port);
    from_port += 1;
  }
  CHECK(port_available);
  return from_port;
}

}  // namespace

class StrategyReporter2 {
 public:
  typedef std::tuple<std::string, std::string, std::string> product_key_type;
  typedef std::tuple<std::string, std::string, std::string> acct_key_type;
  typedef std::tuple<std::string, std::string> market_key_type;
  typedef std::unordered_map<std::string, double> reserve_type;
  typedef std::map<acct_key_type, reserve_type> reserve_map_type;
  typedef std::unordered_map<
      std::string, coin::proto::coin_executor::AggressiveExecutorConfig> executor_type;
  typedef std::unordered_map<std::string, CurrencyBalance> balance_type;
  typedef std::map<acct_key_type, balance_type> balance_map_type;
  typedef std::unordered_map<std::string, coin::proto::ProductPosition> position_type;
  typedef std::map<acct_key_type, position_type> position_map_type;
  typedef std::function<
      void(const coin::proto::StrategyLog&, bool, int64_t)> log_callback_type;

  StrategyReporter2(
      const StrategyManagerConfig& strategy_config)
      : strategy_config_(strategy_config),
        prev_pnl_send_time_(0),
        pnl_interval_(20'000'000'000LL),
        prev_exe_config_send_time_(0),
        exe_config_interval_(600'000'000'000LL) {
    // Strategy Request
    coin::proto::StrategyRequestProto strat_request;
    strat_request.set_strategy_group(strategy_config.strategy_group());
    strat_request.set_strategy_name(strategy_config.strategy_name());
    const auto& strat_slim_logger_config = strategy_config.strategy_slim_logger_config();
    const auto& strat_logger_config = strategy_config.strategy_logger_config();
    const auto& telemetry_logger_config = strategy_config.telemetry_logger_config();
    const auto& health_reporter_config = strategy_config.strategy_health_reporter_config();
    strat_slim_logger_ = std::make_unique<StrategySlimLogger>(
        strat_request,
        strat_slim_logger_config.log_root(),
        strat_slim_logger_config.machine());
    strat_logger_ = std::make_unique<StrategyLogger>(
        strat_request,
        strat_logger_config.log_root(),
        strat_logger_config.machine());
    telemetry_logger_ = std::make_unique<TelemetryLogger>(
        strat_request,
        telemetry_logger_config.log_root(),
        telemetry_logger_config.machine());
    log_callback_type strat_slim_log_write_callback = std::bind(
        &StrategySlimLogger::CondWrite,
        strat_slim_logger_.get(),
        std::placeholders::_1,
        std::placeholders::_2,
        std::placeholders::_3);
    strat_logger_->AddCallback(strat_slim_log_write_callback);
    log_callback_type telemetry_log_write_callback = std::bind(
        &TelemetryLogger::CondWrite,
        telemetry_logger_.get(),
        std::placeholders::_1,
        std::placeholders::_2,
        std::placeholders::_3);
    strat_logger_->AddCallback(telemetry_log_write_callback);

    if (health_reporter_config.has_from_port() &&
        health_reporter_config.has_to_port()) {
      int port = GetAvailablePort(
          health_reporter_config.from_port(), health_reporter_config.to_port());
      health_reporter_ = std::make_unique<PrometheusHealthReporter>(
          fmt::format("0.0.0.0:{}", port));
    }

    auto curr_time = GetCurrentTimestamp();
    strat_logger_->WriteRequest(true, curr_time);

    auto interval = strategy_config.pnl_report_period_ns();
    pnl_interval_ = (interval < pnl_interval_) ? interval : pnl_interval_;
  }

  void RegisterExecutorConfig(int64_t curr_time,
                              const AccountRequestProto& acct_req,
                              const IProduct& product,
                              const PassiveExecutorConfig& executor_config) {
    ValidateAccountRequest(acct_req);
    CHECK_THROW(product.exchange_name() == acct_req.exchange());
    // Bitflyer Spot products are defined as futures products in our code base.
    if (acct_req.exchange() != "Bitflyer") {
      CHECK_THROW(product.market_name() == acct_req.market_type());
    }
    auto acct_key = std::make_tuple(
        acct_req.market_type(), acct_req.exchange(), acct_req.owner());
    accounts_.insert(acct_key);
    if (product.market() == MarketType::Spot) {
      std::string base = product.base().currency();
      std::string quote = product.quote().currency();
      if (spot_acct_quote_.count(acct_key) == 0) {
        spot_acct_quote_[acct_key] = quote;
      } else {
        auto quote_exist = spot_acct_quote_.at(acct_key);
        CHECK_THROW(quote_exist == quote)
          << "spot account quote mismatch." << quote_exist << " vs " << quote;
      }
      auto& acct_reserve = reserve_map_[acct_key];
      std::optional<double> reserve;
      if (executor_config.has_reserve()) {
        reserve = executor_config.reserve();
      } else if (executor_config.has_min_pos() && executor_config.has_max_pos()) {
        reserve = 0.5 * (executor_config.min_pos() + executor_config.max_pos());
      } else {
        THROW() << "unknown reserve";
      }
      acct_reserve[base] = reserve.value();
      if (acct_reserve.count(quote) == 0) {
        acct_reserve[quote] = 0;
      } else {
        CHECK_EQ_THROW(acct_reserve.at(quote), 0);
      }
      UpdateReserveEntry(acct_key, base, reserve.value());
      UpdateReserveEntry(acct_key, quote, 0);
    }
    UpdateExecutorMap(acct_key, product.symbol(), executor_config, curr_time);
  }

  StrategyLogger* GetMutableStrategyLogger() { return strat_logger_.get(); }

  PrometheusHealthReporter* GetMutableHealthReporter() {
      return health_reporter_.get();
  }

  void UpdateMidp(const IProduct& product, std::optional<double> midp) {
    if (product.market() == MarketType::Options &&
        product.exchange() == ExchangeType::Deribit) return;
    UpdatePriceMap(product, midp);
  }

  void UpdateMarkPrice(const IProduct& product, const FeedUpdate& upd) {
    if (product.market() == MarketType::Options &&
        product.exchange() == ExchangeType::Deribit) {
      std::optional<double> mark_price = upd.mark_price().mark_price;
      UpdatePriceMap(product, mark_price);
    }
  }

  void onAccountInfo(const exchange::base::order::OrderUpdate& upd) {
    auto curr_time = GetCurrentTimestamp();
    const auto& owner = upd.account_info().owner();
    auto acct_key = std::make_tuple(MarketType_Name(upd.mea().market),
                                    ExchangeType_Name(upd.mea().exchange),
                                    owner);
    if (accounts_.count(acct_key) == 0) {
      LOG_EVERY_N(ERROR, 10) << GetAccountStr(acct_key) << ": Not Ready Yet";
      return;
    }
    CHECK_EQ_THROW(accounts_.count(acct_key), 1);

    if (upd.mea().market == MarketType::Futures ||
        upd.mea().market == MarketType::Options) {
      auto& acct_pos = position_map_[acct_key];
      for (const auto& it : upd.account_info().GetPositions()) {
        if (!(it.second.has_symbol() && it.second.has_net_position()))
          continue;
        const auto& symbol = it.second.symbol();
        auto& product_pos = acct_pos[symbol];
        product_pos.set_net_position(it.second.net_position());
      }
    }
    auto& acct_bal = balance_map_[acct_key];
    for (const auto& it : upd.account_info().GetBalances()) {
      if (!(it.second.has_currency() && it.second.has_total()))
          continue;
      const auto& currency = it.second.currency();
      auto& currency_bal = acct_bal[currency];
      currency_bal.set_total(it.second.total());
    }

    if (prev_pnl_send_time_ + pnl_interval_ < curr_time) {
      SPDLOG_INFO("####################");
      WriteStrategyLog(curr_time);
      prev_pnl_send_time_ = curr_time;
      SPDLOG_INFO("####################");
    }
  }

 private:
  friend class StrategyReporter2Test;

  const StrategyManagerConfig strategy_config_;
  std::unique_ptr<StrategyLogger> strat_logger_;
  std::unique_ptr<StrategySlimLogger> strat_slim_logger_;
  std::unique_ptr<TelemetryLogger> telemetry_logger_;
  std::unique_ptr<PrometheusHealthReporter> health_reporter_;

  std::map<product_key_type, std::optional<double>> price_map_;

  std::set<acct_key_type> accounts_;
  std::map<acct_key_type, std::string> spot_acct_quote_;
  reserve_map_type reserve_map_;
  std::map<acct_key_type, std::unordered_map<std::string, CurrencyBalance*>> reserve_bal_ptr_;
  std::map<acct_key_type, coin::proto::AccountBalance> acct_balance_for_reserve_;
  std::map<acct_key_type, executor_type> executor_map_;

  balance_map_type balance_map_;
  position_map_type position_map_;
  balance_map_type pnl_bal_buf_;
  int64_t prev_pnl_send_time_;
  int64_t pnl_interval_;
  int64_t prev_exe_config_send_time_;
  int64_t exe_config_interval_;

  std::string GetAccountStr(const acct_key_type& acct_key) const {
    return fmt::format(
        "{} {} {}",
        std::get<0>(acct_key), std::get<1>(acct_key), std::get<2>(acct_key));
  }

  std::string GetMarketStr(const acct_key_type& acct_key) const {
    return fmt::format(
        "{} {}",
        std::get<0>(acct_key), std::get<1>(acct_key));
  }

  void ValidateAccountRequest(const AccountRequestProto& acct_req) {
    CHECK_THROW(acct_req.has_market_type());
    CHECK_THROW(acct_req.has_exchange());
    CHECK_THROW(acct_req.has_owner());
  }

  void UpdatePriceMap(const IProduct& product, std::optional<double> price) {
    auto product_key = std::make_tuple(product.market_name(),
                                       product.exchange_name(),
                                       product.absolute_norm());
    price_map_[product_key] =  price;
    if (price.has_value()) {
      strat_logger_->UpdateMidp(product.order_native_symbol(), price.value());
    }
  }

  void UpdateExecutorMap(const acct_key_type& acct_key,
                         const std::string& symbol,
                         const PassiveExecutorConfig& exe_config,
                         int64_t curr_time) {
    CHECK_THROW(exe_config.has_lot_size());
    CHECK_THROW(exe_config.has_min_pos());
    CHECK_THROW(exe_config.has_max_pos());
    bool executor_config_updated = false;
    if (executor_map_[acct_key].count(symbol) == 0) {
      executor_config_updated = true;
      coin::proto::coin_executor::AggressiveExecutorConfig curr_exe_config;
      curr_exe_config.mutable_account_request()->set_market_type(std::get<0>(acct_key));
      curr_exe_config.mutable_account_request()->set_exchange(std::get<1>(acct_key));
      curr_exe_config.mutable_account_request()->set_owner(std::get<2>(acct_key));
      curr_exe_config.set_symbol(symbol);
      executor_map_[acct_key][symbol] = std::move(curr_exe_config);
    }
    auto& curr_exe_config = executor_map_[acct_key].at(symbol);
    if (!executor_config_updated) {
      if (curr_exe_config.lot_size() != exe_config.lot_size() ||
          curr_exe_config.min_pos() != exe_config.min_pos() ||
          curr_exe_config.max_pos() != exe_config.max_pos() ||
          curr_exe_config.reserve() != exe_config.reserve()) {
        executor_config_updated = true;
      }
    }
    if (executor_config_updated) {
      curr_exe_config.set_lot_size(exe_config.lot_size());
      curr_exe_config.set_min_pos(exe_config.min_pos());
      curr_exe_config.set_max_pos(exe_config.max_pos());
      if (exe_config.has_reserve()) {
        curr_exe_config.set_reserve(exe_config.reserve());
      } else {
        curr_exe_config.clear_reserve();
      }
      strat_logger_->WriteAggressiveExecutorConfig(
          executor_map_.at(acct_key).at(symbol), true, curr_time);
    }
  }

  void UpdateReserveEntry(const acct_key_type& acct_key,
                          const std::string& currency,
                          double reserve) {
    CurrencyBalance* cbal_ptr = nullptr;
    if (reserve_bal_ptr_[acct_key].count(currency) == 0) {
      if (acct_balance_for_reserve_.count(acct_key) == 0) {
        AccountBalance acct_bal;
        acct_bal.set_market_type(std::get<0>(acct_key));
        acct_bal.set_exchange(std::get<1>(acct_key));
        acct_bal.set_owner(std::get<2>(acct_key));
        acct_balance_for_reserve_[acct_key] = acct_bal;
      }
      cbal_ptr = acct_balance_for_reserve_.at(acct_key).add_each_balance();
      reserve_bal_ptr_[acct_key][currency] = cbal_ptr;
    } else {
      cbal_ptr = reserve_bal_ptr_.at(acct_key).at(currency);
    }
    cbal_ptr->set_currency(currency);
    cbal_ptr->set_total(reserve);
  }

  void UpdateSpotPnlBalance(const acct_key_type& acct_key) {
    std::unordered_map<std::string, double> pos_map;
    SPDLOG_INFO("####################");
    const auto& acct_reserve = reserve_map_.at(acct_key);

    bool pnl_balance_ready = true;
    if (balance_map_.count(acct_key) == 0) {
      pnl_balance_ready = false;
    }
    if (pnl_balance_ready) {
      const auto& acct_balance = balance_map_.at(acct_key);
      for (const auto& [currency, reserve] : acct_reserve) {
        if (acct_balance.count(currency) == 0) {
          pnl_balance_ready = false;
          break;
        } else {
          auto total = acct_balance.at(currency).total();
          CHECK_EQ_THROW(pos_map.count(currency), 0);
          SPDLOG_INFO("# {}: {}", currency, total);
          pos_map[currency] = total - reserve;
        }
      }
    }

    const auto& quote = spot_acct_quote_.at(acct_key);
    double pnl_balance_value = 0;
    if (pnl_balance_ready) {
      for (const auto& [currency, pos] : pos_map) {
        std::string symbol = currency + "-" + quote;
        auto product_key = std::make_tuple(
            std::get<0>(acct_key), std::get<1>(acct_key), symbol);
        std::optional<double> midp;
        if (currency == quote) {
          midp = 1;
        } else {
          midp = price_map_[product_key];
        }
        if (!midp.has_value()) {
          pnl_balance_ready = false;
          SPDLOG_WARN("No midp for {} {}", GetMarketStr(acct_key), symbol);
          break;
        }
        double midp_value = midp.value();
        double ccy_pnl_bal = pos * midp.value();
        pnl_balance_value += ccy_pnl_bal;
        SPDLOG_INFO(
            "# {} reserve: {} pos: {} midp: {} pnlbal: {}",
            symbol, reserve_map_.at(acct_key).at(currency), pos, midp_value, ccy_pnl_bal);
      }
    }

    auto& pnl_balance = pnl_bal_buf_[acct_key][quote];
    pnl_balance.Clear();
    if (pnl_balance_ready) {
      pnl_balance.set_total(pnl_balance_value);
    }
  }

  void UpdateFuturesPnlBalance(const acct_key_type& acct_key) {
    if (position_map_.count(acct_key) == 1) {
      SPDLOG_INFO("********************");
      for (const auto& [symbol, position] : position_map_.at(acct_key)) {
        if (position.has_net_position()) {
          SPDLOG_INFO("* {}: {}", symbol, position.net_position());
        }
      }
    }
    SPDLOG_INFO("********************");
    SPDLOG_INFO("####################");
    if (balance_map_.count(acct_key) == 1) {
      const auto& acct_balance = balance_map_.at(acct_key);
      for (const auto& [currency, balance] : acct_balance) {
        auto& pnl_balance = pnl_bal_buf_[acct_key][currency];
        pnl_balance.Clear();
        if (balance.has_total()) {
          SPDLOG_INFO("# {}: {}", currency, balance.total());
          pnl_balance.set_total(balance.total());
        }
      }
    }
  }

  void UpdatePnlBalance() {
    for (const auto& acct_key : accounts_) {
      SPDLOG_INFO("####################");
      SPDLOG_INFO("# {}", GetAccountStr(acct_key));
      const auto& market_type = std::get<0>(acct_key);
      if (market_type == "Spot") {
        UpdateSpotPnlBalance(acct_key);
      } else if (market_type == "Futures" || market_type == "Options") {
        UpdateFuturesPnlBalance(acct_key);
      } else {
        THROW() << "Unknown market type: " << market_type;
      }
    }
  }

  void GetAggPnlBalance(std::unordered_map<std::string, double>* agg_pnl_bal_map,
                        std::vector<AccountBalance>* pnl_balance_details) {
    UpdatePnlBalance();
    agg_pnl_bal_map->clear();
    pnl_balance_details->clear();
    bool ready = true;
    SPDLOG_INFO("####################");
    for (const auto& acct_key : accounts_) {
      bool acct_ready = true;
      if (pnl_bal_buf_.count(acct_key) == 0) {
        acct_ready = false;
      }
      if (acct_ready) {
        const auto& market_type = std::get<0>(acct_key);
        const auto& exchange = std::get<1>(acct_key);
        const auto& owner = std::get<2>(acct_key);
        AccountBalance acct_pnl_bal;
        acct_pnl_bal.set_market_type(market_type);
        acct_pnl_bal.set_exchange(exchange);
        acct_pnl_bal.set_owner(owner);
        for (const auto& [currency, pnl_bal] : pnl_bal_buf_.at(acct_key)) {
          if (!pnl_bal.has_total()) {
            acct_ready = false;
            break;
          }
          if (market_type == "Options" && exchange == "Deribit") {
            auto futures_acct_key = std::make_tuple("Futures", exchange, owner);
            if (accounts_.count(futures_acct_key) != 0) {
              SPDLOG_INFO(
                  "ignore pnl balance of {}.{}.{} to prevent duplicated pnl balance",
                  market_type, exchange, owner);
              break;
            }
          }
          if (agg_pnl_bal_map->count(currency) == 0) {
            (*agg_pnl_bal_map)[currency] = 0;
          }
          agg_pnl_bal_map->at(currency) += pnl_bal.total();
          SPDLOG_INFO("## {} {} pnlbal: {}", GetAccountStr(acct_key), currency, pnl_bal.total());
          auto each_balance = acct_pnl_bal.add_each_balance();
          each_balance->set_currency(currency);
          each_balance->set_total(pnl_bal.total());
        }
        if (acct_pnl_bal.each_balance().size() > 0) {
          pnl_balance_details->push_back(acct_pnl_bal);
        }
      }
      if (!acct_ready) {
        SPDLOG_INFO("pnl balance not ready: {}", GetAccountStr(acct_key));
      }
      ready = ready && acct_ready;
    }
    if (!ready) {
      agg_pnl_bal_map->clear();
      pnl_balance_details->clear();
    }
  }

  void WriteBalance(int64_t curr_time) {
    for (const auto& [acct_key, acct_bal] : balance_map_) {
      AccountBalance acct_bal_proto;
      acct_bal_proto.set_market_type(std::get<0>(acct_key));
      acct_bal_proto.set_exchange(std::get<1>(acct_key));
      acct_bal_proto.set_owner(std::get<2>(acct_key));
      for (const auto& [currency, ccy_bal] : acct_bal) {
        CurrencyBalance* dst_ccy_bal = acct_bal_proto.add_each_balance();
        dst_ccy_bal->set_currency(currency);
        dst_ccy_bal->set_total(ccy_bal.total());
      }
      strat_logger_->WriteBalance(acct_bal_proto, true, curr_time);
    }
  }

  void WriteTicker(int64_t curr_time) {
    std::map<market_key_type, coin::proto::ExchangeTicker> all_ticker;
    for (const auto& [prod_key, midp] : price_map_) {
      std::string market_type = std::get<0>(prod_key);
      std::string exchange = std::get<1>(prod_key);
      std::string symbol = std::get<2>(prod_key);
      if (!midp.has_value())
        continue;
      auto market_key = std::make_tuple(market_type, exchange);
      if (all_ticker.count(market_key) == 0) {
        coin::proto::ExchangeTicker exchange_ticker;
        exchange_ticker.set_market_type(market_type);
        exchange_ticker.set_exchange(exchange);
        all_ticker[market_key] = exchange_ticker;
      }
      auto product_ticker = all_ticker.at(market_key).add_each_ticker();
      product_ticker->set_symbol(symbol);
      product_ticker->set_last(midp.value());
    }
    for (const auto& [market_key, exchange_ticker] : all_ticker) {
      strat_logger_->WriteTicker(exchange_ticker, true, curr_time);
    }
  }

  void WriteStrategyLog(int64_t curr_time) {
    if (strategy_config_.has_reporter_config()) {
      strat_logger_->WriteStratReporterConfig(
          strategy_config_.reporter_config(), true, curr_time);
    }
    for (const auto& [acct_key, acct_reserve] : acct_balance_for_reserve_) {
      if (acct_reserve.each_balance_size() > 0) {
        strat_logger_->WriteReserve(acct_reserve, true, curr_time);
      }
    }
    if (prev_exe_config_send_time_ + exe_config_interval_ < curr_time) {
      prev_exe_config_send_time_ = curr_time;
      for (const auto& [acct_key, acct_executor] : executor_map_) {
        for (const auto& [symbol, exe_config] : acct_executor) {
          strat_logger_->WriteAggressiveExecutorConfig(exe_config, true, curr_time);
        }
      }
    }
    WriteBalance(curr_time);
    WriteTicker(curr_time);
    std::unordered_map<std::string, double> agg_pnl_bal_map;
    std::vector<AccountBalance> pnl_bal_details;
    GetAggPnlBalance(&agg_pnl_bal_map, &pnl_bal_details);
    if (pnl_bal_details.size() > 0) {
      strat_logger_->WritePnlBalanceDetails(pnl_bal_details, true, curr_time);
    }
    for (const auto& [currency, pnl_bal_value] : agg_pnl_bal_map) {
      CurrencyBalance bal;
      bal.set_currency(currency);
      bal.set_total(pnl_bal_value);
      strat_logger_->WritePnlBalance(bal, std::nullopt, true, curr_time);
      SPDLOG_INFO("### Total {} pnlbal: {}", currency, pnl_bal_value);
    }
  }
};

}  // namespace coin2::strategy::util
